What was broken

When you open a task in the UI, the backend keeps a live stream open to push updates. That stream had no way to end itself — it only stopped if the browser disconnected.

So when a task finished, the backend just kept listening forever, holding one Redis connection each time. These pile up until the pool is empty, at which point the pod can't serve requests (and its health check fails, so it sits stuck until a manual restart).

A second bug: when one viewer left, the code deleted the task's shared event stream — yanking it out from under everyone else watching the same task.

The fix

• End the stream when the task is done. The backend already gets a task_updated event when a task finishes, so the stream now closes as soon as it sees a terminal status.
• Fallback for the rare cases the event can't cover (viewer connects after the task already finished, or the producer dies silently): a lightweight check on the existing keepalive interval closes the stream then too.
• Stop deleting the shared stream on one viewer's exit — the Redis TTL already cleans it up on its own.

Testing

Two integration regression tests (stream self-ends on terminal task; one viewer disconnecting doesn't delete the shared stream).

Verified live (A/B against main) — same steps both times: create a RUNNING task → open GET /tasks/{id}/stream → complete the task, while watching Redis blocked_clients.

On main the reader logged Reading messages from Redis stream … every ~2s indefinitely after the task was done; on this branch it stops the moment the terminal event arrives. Same reproduction the incident described, and the connection is released instead of pinned.

Note: the unit/integration suite runs in CI; the tutorial-agent integration jobs were red due to a pre-existing missing-pytest-asyncio issue unrelated to this change (fixed separately in scale-agentex-python).

Related: the UI-side half of this leak is #383.

🤖 Generated with Claude Code

Greptile Summary

This PR updates task SSE subscription lifecycle handling.

• Ends subscriptions when a terminal update is received, authoritative task state becomes terminal, or the task disappears.
• Replays buffered events and closes when connecting to an already-terminal task.
• Publishes failure updates and centralizes terminal/non-terminal status classification.
• Preserves shared Redis streams when individual subscribers disconnect.
• Adds integration coverage for terminal shutdown, late connections, reclaimed keys, transient lookups, and shared-stream retention.

sequenceDiagram
    participant Client
    participant SSE as StreamsUseCase
    participant DB as Task Store
    participant Redis
    Client->>SSE: Subscribe to task stream
    SSE->>Redis: Snapshot stream cursor
    SSE->>DB: Read authoritative task state
    alt Task already terminal
        SSE->>Redis: Replay buffered events
        SSE-->>Client: Events, then close
    else Task is non-terminal
        loop Until terminal, missing, or disconnect
            SSE->>Redis: Read new events
            alt Terminal task_updated received
                SSE-->>Client: Terminal event
                SSE-->>Client: Close stream
            else Status-check interval elapsed
                SSE->>DB: Recheck task
                alt Terminal or missing
                    SSE-->>Client: Close stream
                end
            end
        end
    end
    Note over SSE,Redis: Subscriber exit does not delete the shared stream

_{Reviews (14): Last reviewed commit: "fix(agentex): terminate SSE task streams..." | Re-trigger Greptile}